package cn.fanyamin.jms;

import java.io.Serializable;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;



/**
 * The Class JmsSender.
 *
 * @author walter
 */
public class JmsSender extends AbstractMessageSender  {
    
    /** The logger. */
    private static Log logger = LogFactory.getLog(JmsSender.class);

    /** The recv timeout ms. */
    private int recvTimeoutMS = 5000;

    /** The idle timeout ms. */
    private int idleTimeoutMS = 30000;

    /** The time to live. */
    private long timeToLive = 5000;

    /** The jms template. */
    private JmsTemplate jmsTemplate = new JmsTemplate();

    /** The factory. */
    private PooledConnectionFactory factory;

    /** The is started. */
    private volatile boolean isStarted = false;

    /** The is inited. */
    private volatile boolean isInited = false;
    /**
     * Instantiates a new jms sender.
     */
    public JmsSender() {

    }

    /**
     * Sets the recv timeout ms.
     *
     * @param recvTimeoutMS the new recv timeout ms
     */
    public void setRecvTimeoutMS(int recvTimeoutMS) {
        this.recvTimeoutMS = recvTimeoutMS;
    }

    /**
     * Sets the idle timeout ms.
     *
     * @param idleTimeoutMS the new idle timeout ms
     */
    public void setIdleTimeoutMS(int idleTimeoutMS) {
        this.idleTimeoutMS = idleTimeoutMS;
    }

    /**
     * Sets the time to live.
     *
     * @param timeToLive the new time to live
     */
    public void setTimeToLive(long timeToLive) {
        this.timeToLive = timeToLive;
    }

    /**
     * init it
     */
    public void init() {
        if (isInited) {
            return;
        }
        ActiveMQConnectionFactory factory0 = new ActiveMQConnectionFactory();
        if (StringUtils.isNotEmpty(this.userName)
                && StringUtils.isNotEmpty(this.password)) {
            factory0.setUserName(this.userName);
            factory0.setPassword(this.password);
        }
        factory0.setBrokerURL(connectionUrl);
        this.factory = new PooledConnectionFactory(factory0);
        this.factory.setIdleTimeout(idleTimeoutMS);
        this.jmsTemplate.setConnectionFactory(factory);
        this.jmsTemplate.setReceiveTimeout(recvTimeoutMS);
        // this.jmsTemplate.setExplicitQosEnabled(true);
        this.jmsTemplate.setTimeToLive(timeToLive);
        isInited = true;
    }

    /**
     * @param logMsg log message
     */
    public void send(final Serializable logMsg) {
        this.send(logMsg, this.destination);
    }

    /**
     * @param logMsg log message
     * @param aDest a destination
     */
    public void send(final Serializable logMsg, Destination aDest) {
        assert (isStarted && jmsTemplate != null && aDest != null);

        jmsTemplate.send(aDest, new MessageCreator() {

            public Message createMessage(Session session) throws JMSException {
                ObjectMessage msg = session.createObjectMessage(logMsg);
                return msg;
            }
        });
    }

    /**
     * start it
     */
    public void start() {
        if (isStarted) {
            return;
        }
        factory.start();
        isStarted = true;
    }

    /**
     * stop it
     */
    public void stop() {
        if (!isStarted) {
            return;
        }

        factory.stop();
        isStarted = false;

    }
}
